package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.beans.TrafficPageViewBean;
import com.atguigu.edu.realtime.utils.DateFormatUtil;
import com.atguigu.edu.realtime.utils.MyClickhouseUtil;
import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;


public class DwsTrafficVcChArIsNewPageViewWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        // TODO 2 检查点相关设置
        /*//2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置job取消之后，检查点是否保留
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置两个检查点之间最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop101:8020/edu/ck");
        //2.7 设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");*/
        // TODO 3 从kafka相应主题消费数据
        String pageLogTopic = "dwd_traffic_page_log";
        String ujdTopic = "dwd_traffic_user_jump_detail";
        String uvTopic = "dwd_traffic_unique_visitor_detail";
        String groupId = "dws_traffic_vc_ch_ar_isnew_group";
        DataStreamSource<String> pageLogDS = env.addSource(MyKafkaUtil.getKafkaConsumer(pageLogTopic, groupId));
        DataStreamSource<String> ujdDS = env.addSource(MyKafkaUtil.getKafkaConsumer(ujdTopic, groupId));
        DataStreamSource<String> uvDS = env.addSource(MyKafkaUtil.getKafkaConsumer(uvTopic, groupId));
        //pageLogDS.print(">>>");
        //{"common":{"sc":"1","ar":"3","uid":"1397","os":"Android 10.0","ch":"oppo","is_new":"0","md":"Sumsung Galaxy S20","mid":"mid_398","vc":"v2.1.134","ba":"Sumsung","sid":"1fa9c50c-aa4a-4524-ab50-a1072f8518ec"},
        // "page":{"page_id":"home","during_time":5475},
        // "ts":1668825270377}
        //ujdDS.print("$$$");
        //{"common":{"sc":"1","ar":"23","uid":"3043","os":"Android 11.0","ch":"xiaomi","is_new":"1","md":"Xiaomi Mix2 ","mid":"mid_387","vc":"v2.1.134","ba":"Xiaomi","sid":"29cf20f7-0197-48cd-8dcb-9d46b7de1a45"},
        // "page":{"page_id":"home","during_time":13375},
        // "ts":1668846365227}
        //uvDS.print("###");
        //{"common":{"sc":"2","ar":"6","uid":"75","os":"Android 10.0","ch":"oppo","is_new":"0","md":"Xiaomi 9","mid":"mid_260","vc":"v2.1.134","ba":"Xiaomi","sid":"b8d28f09-1ab9-441b-9a61-baae410675b3"},
        // "page":{"page_id":"home","during_time":7807},
        // "ts":1668703765888}


        // TODO 4 对流中的数据进行转换
        //{"common":{"sc":"1","ar":"3","uid":"1397","os":"Android 10.0","ch":"oppo","is_new":"0","md":"Sumsung Galaxy S20","mid":"mid_398","vc":"v2.1.134","ba":"Sumsung","sid":"1fa9c50c-aa4a-4524-ab50-a1072f8518ec"},
        // "page":{"page_id":"home","during_time":5475},
        // "ts":1668825270377}
        SingleOutputStreamOperator<TrafficPageViewBean> pageLogStatsDS = pageLogDS.map(
                new MapFunction<String, TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        JSONObject common = jsonObject.getJSONObject("common");
                        Long ts = jsonObject.getLong("ts");
                        JSONObject page = jsonObject.getJSONObject("page");
                        TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean(
                                "",
                                "",
                                common.getString("vc"),
                                common.getString("ch"),
                                common.getString("ar"),
                                common.getString("is_new"),
                                0L,
                                StringUtils.isEmpty(page.getString("last_page_id")) ? 1L : 0L,
                                1L,
                                page.getLong("during_time"),
                                0L,
                                ts
                        );
                        return trafficPageViewBean;
                    }
                }
        );

        //{"common":{"sc":"1","ar":"23","uid":"3043","os":"Android 11.0","ch":"xiaomi","is_new":"1","md":"Xiaomi Mix2 ","mid":"mid_387","vc":"v2.1.134","ba":"Xiaomi","sid":"29cf20f7-0197-48cd-8dcb-9d46b7de1a45"},
        // "page":{"page_id":"home","during_time":13375},
        // "ts":1668846365227}
        SingleOutputStreamOperator<TrafficPageViewBean> ujdStatsDS = ujdDS.map(
                new MapFunction<String, TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        JSONObject common = jsonObject.getJSONObject("common");
                        Long ts = jsonObject.getLong("ts");
                        JSONObject page = jsonObject.getJSONObject("page");
                        TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean(
                                "",
                                "",
                                common.getString("vc"),
                                common.getString("ch"),
                                common.getString("ar"),
                                common.getString("is_new"),
                                0L,
                                0L,
                                0L,
                                0L,
                                1L,
                                ts
                        );
                        return trafficPageViewBean;
                    }
                }
        );
        SingleOutputStreamOperator<TrafficPageViewBean> uvStatsDS = uvDS.map(
                new MapFunction<String, TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        JSONObject common = jsonObject.getJSONObject("common");
                        Long ts = jsonObject.getLong("ts");
                        JSONObject page = jsonObject.getJSONObject("page");
                        TrafficPageViewBean trafficPageViewBean = new TrafficPageViewBean(
                                "",
                                "",
                                common.getString("vc"),
                                common.getString("ch"),
                                common.getString("ar"),
                                common.getString("is_new"),
                                1L,
                                0L,
                                0L,
                                0L,
                                0L,
                                ts
                        );
                        return trafficPageViewBean;
                    }
                }
        );
        // TODO 5 将三条流关联起来
        DataStream<TrafficPageViewBean> unionDS = pageLogStatsDS
                .union(ujdStatsDS)
                .union(uvStatsDS);

        // TODO 6 指定水位线 及 指定事件时间字段
        SingleOutputStreamOperator<TrafficPageViewBean> withWatermarkDS = unionDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
                            @Override
                            public long extractTimestamp(TrafficPageViewBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })
        );
        // TODO 7 按照维度分组
        KeyedStream<TrafficPageViewBean, Tuple4<String, String, String, String>> keyedDS = withWatermarkDS.keyBy(
                new KeySelector<TrafficPageViewBean, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(TrafficPageViewBean value) throws Exception {
                        return Tuple4.of(
                                value.getVc(),
                                value.getAr(),
                                value.getCh(),
                                value.getIsNew()
                        );
                    }
                }
        );
        // TODO 8 开窗
        WindowedStream<TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow> windowDS
                = keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        // TODO 9 聚合
        SingleOutputStreamOperator<TrafficPageViewBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                        value1.setUjCt(value1.getUjCt() + value2.getUjCt());
                        return value1;
                    }
                }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow window, Iterable<TrafficPageViewBean> input, Collector<TrafficPageViewBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());
                        for (TrafficPageViewBean trafficPageViewBean : input) {
                            trafficPageViewBean.setStt(stt);
                            trafficPageViewBean.setEdt(edt);
                            trafficPageViewBean.setTs(System.currentTimeMillis());
                            out.collect(trafficPageViewBean);
                        }
                    }
                }
        );
        reduceDS.print(">>>>");
        //TODO 10 将聚合的结果写到CK
        reduceDS.addSink(
                MyClickhouseUtil.getSinkFunction("insert into dws_traffic_vc_ch_ar_is_new_page_view_window values(?,?,?,?,?,?,?,?,?,?,?,?)")
        );
        env.execute();
    }

}
